package com.lyf.language.practice.practice_6_21.io.network.tcpv2;

import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.ServerSocket;
import java.net.Socket;
import java.nio.charset.StandardCharsets;
import java.util.Scanner;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Condition;
import java.util.concurrent.locks.ReentrantLock;

/**
 * TODO 并行处理--
 * 结合MQ进行学习！！
 * NIO非阻塞--？
 * @author LYF
 * @date 2022/6/22
 */
public class Server1 {
    /**
     * ThreadPool
     */
    final static ThreadPoolExecutor executor =
            new ThreadPoolExecutor(2,5,2,
                    TimeUnit.SECONDS,new LinkedBlockingQueue<>());

    /*
    锁
     */
    static ReentrantLock reentrantLock = new ReentrantLock();
    static Condition condition1 = reentrantLock.newCondition();
    static Condition condition2 = reentrantLock.newCondition();

    static CountDownLatch countDownLatch = new CountDownLatch(1);


    /**
     * Server
     */
    static void server(){
        ServerSocket server =null;
        AtomicReference<InputStream> inputStream = null;
        AtomicReference<OutputStream> outputStream = null;
        Scanner scanner = new Scanner(System.in);
        try {
            server = new ServerSocket(8888);
            System.out.println("waiting connect...");
            Socket socket = server.accept();//监听socket连接
            countDownLatch.countDown();

            //reentrantLock.unlock();

            System.out.println("connect successfully...");
            // 监听消息
//            while (true){

            executor.submit(()->{
                countDownLatch.await();
                while(true){
                    System.out.println("test1");
                    //assert false;

//                    reentrantLock.lock();
//                    condition1.await();

                    System.out.println("::"+socket.getInputStream());
                    inputStream.set(socket.getInputStream());
                    byte[]bytes = new byte[1024];
                    int readN = inputStream.get().read(bytes);//暂定可一次读取完毕
                    System.out.println("receive from client-"+socket.getRemoteSocketAddress()+"::"+new String(bytes,0,readN));

                }
            });

            executor.submit(()->{
                countDownLatch.await();
                while (true){
                    System.out.println("test2");
                    //assert false;
//                    reentrantLock.lock();
//                    condition2.await();
                    System.out.println(":::"+socket.getOutputStream());
                    outputStream.set(socket.getOutputStream());
                    System.out.println("reply:");
                    outputStream.get().write(scanner.nextLine().getBytes(StandardCharsets.UTF_8));
                }
            });

//            condition1.signal();
//            reentrantLock.unlock();
//            condition2.signal();
//            reentrantLock.unlock();

//            condition2.signal();

//            }

            } catch (IOException e) {
                e.printStackTrace();
            }finally {
//            if(server!=null) {
//                try {
//                    server.close();
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            }
//            assert inputStream != null;
//            if(inputStream.get() !=null){
//                try {
//                    inputStream.get().close();
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            }
//            if(outputStream.get() !=null){
//                try {
//                    outputStream.get().close();
//                } catch (IOException e) {
//                    e.printStackTrace();
//                }
//            }
        }
    }

    public static void main(String[] args) {
        server();
    }

}
